package com.rpcservice.rpc.registry.zk.util;

import com.rpcservice.rpc.common.enums.RpcConfigEnum;
import com.rpcservice.rpc.common.utils.PropertiesFileUtil;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

/**
 * ZooKeeper Java 客户端 Curtor 的工具类
 * 用于创建 ZooKeeper 客户端、注册服务、发现服务，实现分布式锁等操作
 */
public class CuratorUtils {
    private static final Logger logger = Logger.getLogger(CuratorUtils.class.getName());


    private static final int BASE_SLEEP_TIME = 1000;
    private static final int MAX_RETRIES = 3;
    private static final String DEFAULT_ZK_ADDRESS = "115.190.176.43:2181";
    private static final int CONNECTION_TIMEOUT_MS = 5000;
    private static final int SESSION_TIMEOUT_MS = 50000;
    private static final String AUTH_INFO = "zookeeper:hk36.9opklm";


    private static final String ZK_REGISTRY_PATH = "/my-rpc";
    // 存储服务名称和地址的映射关系
    private static final Map<String, List<String>> SERVICE_ADDRESS_MAP = new ConcurrentHashMap<>();
    // 存储注册的路径
    private static final Set<String> REGISTERED_PATH_SET = ConcurrentHashMap.newKeySet();
    // 存储当前连接的 ZooKeeper 客户端,应该需要单例模式
    private static CuratorFramework zkClient;

    /**
     * 私有构造方法，防止实例化
     */
    private CuratorUtils() {
        // 防止实例化
    }

    /**
     * Create persistent nodes. Unlike temporary nodes, persistent nodes are not removed when the client disconnects
     *
     * @param path node path
     */
    public static void CreatePersistentNode(CuratorFramework zkClient, String path) throws Exception {
        try {
            if (REGISTERED_PATH_SET.contains(path) || zkClient.checkExists().forPath(path) != null) {
                logger.info("The node already exists. The node is: [ " + path + "]");
            } else {
                //eg: /my-rpc/com.rpcservice.HelloService/127.0.0.1:9999
                zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
                logger.info("The node created successfully. The node is: [ " + path + "]");
            }
        } catch (Exception e) {
            logger.warning("create persistent node for path [ " + path + " ] failed");
        }
    }

    /**
     * Gets the children under a node
     *
     * @param rpcServiceName rpc service name eg:github.javaguide.HelloServicetest2version1
     * @return All child nodes under the specified node
     */
    public static List<String> getChildrenNodes(CuratorFramework zkClient, String rpcServiceName) throws Exception {
        if (SERVICE_ADDRESS_MAP.containsKey(rpcServiceName)) {
            return SERVICE_ADDRESS_MAP.get(rpcServiceName);
        }
        List<String> result = null;
        String servicePath = ZK_REGISTRY_PATH + "/" + rpcServiceName;
        try {
            result = zkClient.getChildren().forPath(servicePath);

            SERVICE_ADDRESS_MAP.put(rpcServiceName, result);

            registerWatcher(rpcServiceName, zkClient);
        } catch (Exception e) {
            logger.warning("getChildrenNodes for service [ " + rpcServiceName + " ] failed");
        }
        return result;
    }

    /**
     * Empty the registry of data
     */
    public static void clearRegistry(CuratorFramework zkClient, InetSocketAddress inetSocketAddress) throws Exception {
        REGISTERED_PATH_SET.stream().parallel().forEach(p -> {
            try {
                if (p.endsWith(inetSocketAddress.toString())) {
                    zkClient.delete().forPath(p);
                }
            } catch (Exception e) {
                logger.warning("clearRegistry for service [ " + p + " ] failed");
            }
        });
        logger.info("All registered services on the server [ " + REGISTERED_PATH_SET + " ] have been cleared!");
    }

    /**
     * 获取 ZooKeeper 客户端
     *
     * @return CuratorFramework
     */
    public static CuratorFramework getZkClient() {
        //如果用户设置了 zk 地址，请检查
        Properties properties = PropertiesFileUtil.readPropertiesFile(RpcConfigEnum.RPC_CONFIG_PATH.getPropertyValue());
        String zookeeperAddress = properties != null && properties.getProperty(RpcConfigEnum.ZK_ADDRESS.getPropertyValue()) != null ?
                properties.getProperty(RpcConfigEnum.ZK_ADDRESS.getPropertyValue()) : DEFAULT_ZK_ADDRESS;
        //如果 zkClient 已经启动，则直接返回
        if (zkClient != null && zkClient.getState() == CuratorFrameworkState.STARTED) {
            return zkClient;
        }
        // Retry strategy. Retry 3 times, and will increase the sleep time between retries.
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE_SLEEP_TIME, MAX_RETRIES);
        zkClient = CuratorFrameworkFactory.builder()
                .connectString(zookeeperAddress)
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(SESSION_TIMEOUT_MS)
                .connectionTimeoutMs(CONNECTION_TIMEOUT_MS)
                .authorization("digest", AUTH_INFO.getBytes(StandardCharsets.UTF_8))
                .build();
        zkClient.start();
        try {
            // 等待 30 秒，直到连接到 the zookeeper
            if (!zkClient.blockUntilConnected(30, TimeUnit.SECONDS)) {
                throw new RuntimeException("Time out waiting to connect to ZK!");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return zkClient;
    }

    /**
     * Registers to listen for changes to the specified node
     * 用于侦听对指定节点的更改
     * 当指定节点发生变化时，会触发回调函数，从而更新本地缓存中的服务地址列表
     *
     * @param rpcServiceName rpc service name eg:github.javaguide.HelloServicetest2version
     */
    private static void registerWatcher(String rpcServiceName, CuratorFramework zkClient) throws Exception {
        String servicePath = ZK_REGISTRY_PATH + "/" + rpcServiceName;
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, servicePath, true);
        PathChildrenCacheListener listener = (CuratorFramework, pathChildrenCacheEvent) -> {
            List<String> serviceAddresses = CuratorFramework.getChildren().forPath(servicePath);
            SERVICE_ADDRESS_MAP.put(rpcServiceName, serviceAddresses);
        };
        pathChildrenCache.getListenable().addListener(listener);
        pathChildrenCache.start();
    }

    /**
     * create ephemeralNode
     *
     * @param zkClient
     * @param path
     * @throws Exception
     */
    public static void createEphemeralNode(CuratorFramework zkClient, String path) throws Exception {
        try {
            //创建临时节点（EPHEMERAL）
            String createdPath = zkClient.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)//设置权限，这里是开放所有权限
                    .forPath(path, "status:ok".getBytes());
        } catch (Exception e) {
            logger.info("create ephemeral node for path [ " + path + " ] failed");
        }
    }

    public static void deleteEphemeralNode(CuratorFramework zkClient, String path) throws Exception {
        //强制删除节点，无论是否有子节点
        try {
            zkClient.delete().deletingChildrenIfNeeded().forPath(path);
        } catch (Exception e) {
            logger.info("delete ephemeral node for path [ " + path + " ] failed");
        }
    }

}
